package com.sync.mutiParse;

import java.nio.ByteBuffer;

import com.alibaba.middleware.race.sync.Server;
import com.constants.Base;
import com.constants.Column;
import com.constants.Constants;
import com.koloboke.collect.map.hash.HashIntIntMap;
import com.sync.utils.Message;

/**
 * 函数入口
 * 
 * @author LittleBearWinnie
 *
 */
public class MutilSyncDemoMain extends Base {

	static final ReplayTask[] replayTasks = new ReplayTask[Constants.ReplayThreadNum];
	static final PaserTask[] paserTasks = new PaserTask[Constants.PaserThreadNum];
	static final byte[] dataBytes = new byte[Constants.end - Constants.start
			+ 1];
	public static void main(String[] args) throws Exception {
		long start = System.currentTimeMillis();
		working(false);
		byte[] data = reBuild();
		ResultWrite.writeToFile(data);
		loggerServer.info("总耗时:" + (System.currentTimeMillis() - start));
	}

	public static void working(boolean bad) throws Exception {
		ReadManager.init();
		for (int i = 0; i < Constants.PaserThreadNum; ++i) {
			paserTasks[i] = new PaserTask(dataBytes, bad);
			paserTasks[i].start();
		}
		for (int i = 0; i < Constants.ReplayThreadNum; ++i) {
			replayTasks[i] = new ReplayTask(i);
			replayTasks[i].start();
		}

		for (int i = 0; i < Constants.ReplayThreadNum; ++i) {
			replayTasks[i].join();
		}
		for (int i = 0; i < Constants.PaserThreadNum; ++i) {
			paserTasks[i].join();
		}
	}

	public static byte[] reBuild() throws Exception {
	    loggerServer.info("重放最后数据，时间 + " + (System.nanoTime()/1000000));
		ByteBuffer resBuffer = ByteBuffer.allocate(40 * 1024 * 1024);
		int pos;
		byte tmpByte;
		HashIntIntMap indexMap;
		ByteBuffer middleBuffer;
		for (int replayPK = Constants.start; replayPK < Constants.end; replayPK++) {
			if (dataBytes[replayPK - Constants.start] != 1) {
				continue;
			}
			// int replayPK = start + 1; replayPK != end; ++replayPK
			int index = replayPK & Constants.mod;
			indexMap = replayTasks[index].indexMap;
			pos = indexMap.get(replayPK);
			if (pos == 0) {
				continue;
			}
			middleBuffer = replayTasks[index].middleBuffer;
			resBuffer.put(Integer.toString(replayPK).getBytes(charsetName));// 优化
			resBuffer.put(TAB);
			for (int filed = 0; filed < Column.count; filed++) {
				middleBuffer.position(pos + filed * Column.chunk);
				while ((tmpByte = middleBuffer.get()) != SEP) {
					resBuffer.put(tmpByte);
				}
				resBuffer.put(Constants.TAB);
			}
			resBuffer.put(resBuffer.position() - 1, SEPLINE);

		}
		byte[] data = new byte[resBuffer.position()];
		resBuffer.flip();
		resBuffer.get(data);
        loggerServer.info("重放最后数据结束，时间 + " + (System.nanoTime()/1000000));
		return data;
	}

	public static byte[] reBuild2() throws Exception {
		loggerServer.info("重放最后数据，时间 + " + (System.nanoTime()/1000000));
		ByteBuffer resBuffer = ByteBuffer.allocate(40 * 1024 * 1024);
		int pos;
		byte tmpByte;
		HashIntIntMap indexMap;
		ByteBuffer middleBuffer;
		for (int replayPK = Constants.start; replayPK < Constants.end; replayPK++) {
			// int replayPK = start + 1; replayPK != end; ++replayPK
			int index = replayPK & Constants.mod;
			indexMap = replayTasks[index].indexMap;
			pos = indexMap.get(replayPK);
			if (pos == 0) {
				continue;
			}
			middleBuffer = replayTasks[index].middleBuffer;
			resBuffer.put(Integer.toString(replayPK).getBytes(charsetName));// 优化
			resBuffer.put(TAB);
			for (int filed = 0; filed < Column.count; filed++) {
				middleBuffer.position(pos + filed * Column.chunk);
				while ((tmpByte = middleBuffer.get()) != SEP) {
					resBuffer.put(tmpByte);
				}
				resBuffer.put(Constants.TAB);
			}
			resBuffer.put(resBuffer.position() - 1, SEPLINE);

		}
		byte[] data = new byte[resBuffer.position()];
		resBuffer.flip();
		resBuffer.get(data);
		loggerServer.info("重放最后数据结束，时间 + " + (System.nanoTime()/1000000));
		return data;
	}

	public static void send(byte[] data) {
		Message message = new Message();
		message.setStatus(Message.ask);
		message.setData(data);
		Server.channelMap.get("client").writeAndFlush(message);
	}

}
